package com.example.artemis.service;

import com.example.artemis.config.QueueProducerConfig;
import com.example.artemis.factory.QueueMessageProducerFactory;
import com.example.artemis.interfaces.MessageProducer;
import com.example.artemis.model.QueueMessage;
import com.example.artemis.statistics.StatisticsService;
import com.example.artemis.statistics.DefaultStatisticsService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 重构后的消息生产者服务
 * 负责消息生成和发送的协调管理
 * 统计功能已解耦到独立的统计服务中
 */
public class MessageProducerService {

    private static final Logger logger = LoggerFactory.getLogger(MessageProducerService.class);

    private final MessageProducer messageProducer;
    private final QueueProducerConfig config;
    private final StatisticsService statisticsService;
    private volatile boolean running = false;
    private ExecutorService executorService;

    public MessageProducerService() {
        this(new DefaultStatisticsService());
    }

    public MessageProducerService(StatisticsService statisticsService) {
        // 创建配置
        this.config = new QueueProducerConfig();

        // 创建工厂
        QueueMessageProducerFactory factory = new QueueMessageProducerFactory();

        // 创建生产者实例
        this.messageProducer = factory.createProducer(config);
        this.statisticsService = statisticsService;

        // 不在这里启动统计服务，由外部统一管理
        logger.debug("MessageProducerService 初始化完成，使用外部统计服务");
    }

    /**
     * 启动生产者服务并开始生成和发送消息
     */
    public CompletableFuture<ProductionResult> startProducers() {
        logger.info("启动消息生产者服务");
        running = true;

        return CompletableFuture.runAsync(this::startMessageProduction)
                .thenApply(v -> {
                    // 现在统计由统计服务管理，这里返回基本的结果信息
                    return new ProductionResult(0, 0); // 实际数据由统计服务提供
                });
    }

    /**
     * 开始消息生产
     */
    private void startMessageProduction() {
        int producerCount = config.getProducerCount();
        int messagesPerProducer = config.getMessagesPerProducer();

        logger.info("启动 {} 个生产者，每个发送 {} 条消息", producerCount, messagesPerProducer);

        // 创建线程池用于并发发送消息
        executorService = Executors.newFixedThreadPool(producerCount);

        // 启动每个生产者线程
        for (int i = 0; i < producerCount; i++) {
            final int producerIndex = i;
            executorService.submit(() -> {
                Thread.currentThread().setName(config.getProducerNamePrefix() + "-" + producerIndex);
                try {
                    sendBatchMessages(producerIndex, messagesPerProducer);
                } catch (Exception e) {
                    logger.error("生产者线程发生错误", e);
                }
            });
        }

        // 等待所有生产者完成
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                logger.warn("生产者线程池未能在指定时间内完成");
                executorService.shutdownNow();
            }
            logger.info("所有生产者已完成消息发送");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            logger.error("等待生产者完成时被中断");
            executorService.shutdownNow();
        }
    }

    /**
     * 批量发送消息
     */
    private void sendBatchMessages(int producerIndex, int messageCount) {
        String producerName = config.getProducerNamePrefix() + "-" + producerIndex;

        logger.info("生产者 {} 开始发送 {} 条消息", producerName, messageCount);

        try {
            for (int i = 0; i < messageCount && running; i++) {
                try {
                    // 生成消息对象
                    QueueMessage message = createQueueMessage(i, producerIndex);

                    // 计算队列名称
                    String queueName = calculateQueueName(message);

                    // 发送消息
                    sendMessage(queueName, message);

                    // 发送间隔
                    if (config.getMessageIntervalMs() > 0) {
                        Thread.sleep(config.getMessageIntervalMs());
                    }

                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    logger.info("生产者 {} 被中断", producerName);
                    break;
                } catch (Exception e) {
                    logger.error("生产者 {} 发送消息时发生错误: {}", producerName, e.getMessage(), e);
                    // 继续尝试下一条消息，不中断整个批次
                }
            }
        } finally {
            // 确保线程结束时清理JMS资源
            try {
                if (messageProducer instanceof com.example.artemis.abstracts.AbstractMessageProducer) {
                    ((com.example.artemis.abstracts.AbstractMessageProducer) messageProducer).closeCurrentThreadResources();
                }
            } catch (Exception e) {
                logger.debug("清理线程JMS资源时发生错误", e);
            }
        }

        logger.info("生产者 {} 完成消息发送", producerName);
    }

    /**
     * 创建Queue消息对象
     */
    private QueueMessage createQueueMessage(int sequence, int producerIndex) {
        QueueMessage message = new QueueMessage();
        message.setId("msg-" + producerIndex + "-" + sequence);
        message.setContent("消息内容来自生产者" + producerIndex + "，序号：" + sequence);
        message.setProducerId("producer-" + producerIndex);
        message.setCreatedTime(System.currentTimeMillis());
        return message;
    }

    /**
     * 根据消息计算队列名称
     */
    private String calculateQueueName(QueueMessage message) {
        int queueIndex = Math.abs(message.hashCode() % config.getQueueCount());
        return config.getQueueNamePrefix() + "." + queueIndex;
    }

    /**
     * 发送消息
     */
    private void sendMessage(String queueName, Serializable message) throws Exception {
        try {
            // 调用生产者发送消息（统计事件由抽象类中的事件发布处理）
            messageProducer.sendMessage(queueName, message);
            logger.debug("消息发送成功 - 队列: {}", queueName);
        } catch (Exception e) {
            logger.error("消息发送失败 - 队列: {}, 错误: {}", queueName, e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 停止生产者服务
     */
    public void stop() {
        logger.info("停止消息生产者服务");
        running = false;

        if (executorService != null && !executorService.isShutdown()) {
            executorService.shutdownNow();
            try {
                if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                    logger.warn("强制关闭生产者线程池");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        // 关闭生产者资源
        if (messageProducer != null) {
            try {
                // 使用反射调用close方法，因为接口中没有定义close方法
                if (messageProducer.getClass().getMethod("close") != null) {
                    messageProducer.getClass().getMethod("close").invoke(messageProducer);
                }
            } catch (Exception e) {
                logger.debug("关闭生产者资源时发生错误", e);
            }
        }

        // 不在这里停止统计服务，由外部统一管理
        logger.info("消息生产者服务已停止");
    }

    /**
     * 获取生产统计信息
     */
    public ProductionStats getStats() {
        if (statisticsService != null) {
            var globalStats = statisticsService.getGlobalStatistics();
            return new ProductionStats(
                globalStats.getTotalSent(),
                globalStats.getTotalFailed(),
                globalStats.getTotalRetried(),
                globalStats.getAverageProductionTime()
            );
        }
        return new ProductionStats(0, 0, 0, 0);
    }

    /**
     * 获取统计服务实例
     */
    public StatisticsService getStatisticsService() {
        return statisticsService;
    }

    /**
     * 检查服务是否正在运行
     */
    public boolean isRunning() {
        return running;
    }

    /**
     * 获取生产进度信息
     */
    public String getProductionProgress() {
        long totalTarget = (long) config.getProducerCount() * config.getMessagesPerProducer();
        long currentSent = 0;

        if (statisticsService != null) {
            currentSent = statisticsService.getGlobalStatistics().getTotalSent();
        }

        double progress = totalTarget > 0 ? (double) currentSent / totalTarget * 100 : 0;

        return String.format("生产进度: %d/%d (%.1f%%)", currentSent, totalTarget, progress);
    }

    // ================== 数据类定义 ==================

    /**
     * 生产结果
     */
    public record ProductionResult(long totalMessagesSent, long totalMessagesFailed) {}

    /**
     * 生产统计信息
     */
    public record ProductionStats(
        long totalSent,
        long totalFailed,
        long totalRetried,
        double averageProductionTime
    ) {}
}